/**
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
 * agreements.  See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership.  The ASF licenses this file to you under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with the License.  You may obtain
 * a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */

package org.apache.zookeeper.server.quorum;

import org.apache.jute.BinaryOutputArchive;
import org.apache.zookeeper.common.Time;
import org.apache.zookeeper.server.FinalRequestProcessor;
import org.apache.zookeeper.server.Request;
import org.apache.zookeeper.server.RequestProcessor;
import org.apache.zookeeper.server.SyncRequestProcessor;
import org.apache.zookeeper.server.ZooKeeperThread;
import org.apache.zookeeper.server.quorum.QuorumPeer.LearnerType;
import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier;
import org.apache.zookeeper.server.util.ZxidUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.net.BindException;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketAddress;
import java.net.SocketException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicLong;

import javax.security.sasl.SaslException;

/**
 * This class has the control logic for the Leader.
 */
public class Leader {
    /**
     * 这个消息是Follower收到的第一个Leader发送的消息。
     * 这个消息包含Leader服务器的协议版本（protocol version） 和 epoch
     *
     * todo: Leader选举之后，各角色服务器启动流程：
     * 1、Leader服务器使用 LearnerCnxAcceptor 接收所有Learner服务器socket连接请求后，新建一个 LearnerHandler 并启动线程。
     *  这个线程主要是Leader服务器和每个Learner服务器之间网络socket通信，并处理数据包使用的。
     * 2、Leader服务器
     */
    public static final int LEADERINFO = 17;
    /**
     * Learner -> Leader
     * Learner在接收到 Leader服务器发来的 LEADERINFO 消息后，会将自己最新的 zxid和epoch以 ACKEPOCH 消息的形式发送给Leader服务器
     */
    public static final int ACKEPOCH = 18;
    /**
     * Leader -> Follower
     * 该消息ZooKeeper实现ZAB算法的核心所在，即ZAB协议中的提议。
     * 在处理事务请求的时候，Leader服务器会将事务请求以PROPOSAL消息的形式创建投票发送给集群中的所有Follower服务器来进行事务日志的记录。
     */
    public final static int PROPOSAL = 2;
    /**
     * Leader -> Learner
     * 用于通知Learner服务器，Leader即将与其进行 DIFF fangshi 的数据同步。
     */
    final static int DIFF = 13;
    /**
     *
     * Leader -> Learner
     * 用于触发Learner服务器进行内存数据库的回滚
     */
    final static int TRUNC = 14;
    /**
     * Leader -> Learner
     * 用于通知Learner服务器，Leader即将与其进行 全量 方式的数据同步。
     */
    final static int SNAP = 15;
    /**
     * Observer -> Leader
     * 类似{@link this#FOLLOWERINFO}
     */
    final static int OBSERVERINFO = 16;
    /**
     * This message type is sent by the leader to indicate(表明; 显示; 象征; 暗示; 间接提及; 示意) it's zxid and if needed, its database.
     *
     * {@link Leader#lead}方法中，Leader开始管理集群：
     * Leader提议出一个Epoch和zxid之后，发送 NEWLEADER 消息，询问所有的Follower是否同意此Epoch和zxid
     */
    final static int NEWLEADER = 10;
    /**
     * Follower服务器 -> Leader服务器
     * Follower服务器发送 lastZxid 给Leader服务器，
     */
    final static int FOLLOWERINFO = 11;
    /**
     * Leader -> Learner
     * 用来告诉Learner服务器，已经完成了数据同步，可以开始对外提供服务了。
     * This message type is sent by the leader to indicate that the follower is
     * now uptodate andt can start responding to clients.
     */
    final static int UPTODATE = 12;
    /**
     * Learner -> Leader
     * 当Learner服务器接收到客户端的事务请求后，就会将请求以 REQUEST 消息的形式转发给Leader服务器处理。
     * This message type is sent to a leader to request and mutation operation.
     * The payload will consist of a request header followed by a request.
     */
    final static int REQUEST = 1;
    /**
     * Follower服务器 -> Leader服务器
     * Follower‘同步’完成一个提议之后，发送这个类别的消息。
     *
     * 注释：‘同步’指 {@link SyncRequestProcessor}请求处理器 把事务请求写到事务日志中。
     */
    final static int ACK = 3;
    /**
     * Leader服务器 -> Follower服务器
     * 通知集群中的所有Follower服务器惊醒事务请求的提交。
     * Leader服务器在接收到过半的Follower服务器发来的 ACK 消息后，就进入事务请求的最终提交流程。
     */
    final static int COMMIT = 4;
    /**
     * Leader服务器 -> Learner服务器
     * 用于Leader服务器同步Learner服务器上的客户端心跳检测，用来激活存活的客户端。
     *
     * 描述：zk客户端会随机与任意一个zk服务器保持连接，因此Leader服务器无法直接接收到所有client的心跳检测，需要委托给Learner服务器保存client的心跳检测记录。
     * Leader服务器会定时向Learner服务器发送PING消息，Learner服务器在接收到PING消息后，会将这段事件内保持心跳检测的client列表，同样以PING消息形式反馈该Leader服务器。
     * 由Leader服务器来负责逐个对这些client进行会话激活。
     * todo:如果Leader服务器与Learner服务器之间超过一定时间没有接收到PING消息，会修改服务器状态吗？
     */
    final static int PING = 5;
    /**
     * Learner服务器 -> Leader服务器
     * 用于Learner服务器校验会话是否有效，同时也会激活会话。
     * 这通常发生在客户端重连的过程中，新的服务器需要向Leader服务器发送此消息已确定该回话是否已经超时。
     */
    final static int REVALIDATE = 6;
    /**
     * Leader -> Learner
     * 用于通知Learner服务器已经完成了 数据同步 操作。
     */
    final static int SYNC = 7;
    /**
     * Learner -> Observer
     * 只有Observer才会收到此类型的packet消息。
     * 对这个消息的处理相当于 PROPOSAL 和 COMMMIT.
     *
     * 描述：在事务请求提交阶段，针对Follower，Leader仅仅只需要发送一个commit消息，Follower就可以完成事务请求的提交了。
     * 因为在这之前的事务请求投票阶段，Follower已经接受过 proposal消息，该消息中包含了事务请求的内容。因此Follower可以从改之前的proposal缓存中再次获取到事务请求。
     * 
     * 而对于Observer来说，由于之前没有参与事务请求的投票，因此没有该事务请求的上下文，显然，Observer无法完成事务请求的提交的，
     * 为了解决这个问题，zk特别设计了 INFORM消息，该消息不仅能够通知Observer已经可以提交事务请求，同时还会在消息中携带事务请求的内容。
     */
    final static int INFORM = 8;

    private static final Logger LOG = LoggerFactory.getLogger(Leader.class);
    static final private boolean nodelay = System.getProperty("leader.nodelay", "true").equals("true");

    static {
        LOG.info("TCP NoDelay set to: " + nodelay);
    }

    final LeaderZooKeeperServer zk;
    final QuorumPeer self;
    //Follower counter
    final AtomicLong followerCounter = new AtomicLong(-1);
    // list of all the followers
    private final HashSet<LearnerHandler> learners = new HashSet<LearnerHandler>();
    // list of followers that are ready to follow (i.e synced with the leader)
    private final HashSet<LearnerHandler> forwardingFollowers = new HashSet<LearnerHandler>();
    private final HashSet<LearnerHandler> observingLearners = new HashSet<LearnerHandler>();
    // Pending sync requests. Must access under 'this' lock.
    private final HashMap<Long, List<LearnerSyncRequest>> pendingSyncs = new HashMap<Long, List<LearnerSyncRequest>>();
    // the follower acceptor thread
    LearnerCnxAcceptor cnxAcceptor;
    ServerSocket ss;
    ConcurrentMap<Long, Proposal> outstandingProposals = new ConcurrentHashMap<Long, Proposal>();
    ConcurrentLinkedQueue<Proposal> toBeApplied = new ConcurrentLinkedQueue<Proposal>();
    Proposal newLeaderProposal = new Proposal();
    StateSummary leaderStateSummary;
    long epoch = -1;
    volatile boolean readyToStart = false;
    boolean isShutdown;
    long lastCommitted = -1;
    /**
     * 最后被提议的zxid
     *
     * 此变量的赋值在：
     *  1、Leader刚开始lead时。{@link Leader#lead()}
     *
     */
    long lastProposed;
    private boolean quorumFormed = false;

    /**
     * 当前集群状态是否在等待确认new epoch状态。
     */
    boolean waitingForNewEpoch = true;

    /**
     * 已经启动并与Leader建立连接的服务器，包含Leader和Learner
     */
    private HashSet<Long> connectingFollowers = new HashSet<Long>();

    private boolean electionFinished = false;

    /**
     * sid
     */
    private HashSet<Long> electingFollowers = new HashSet<Long>();

    Leader(QuorumPeer self, LeaderZooKeeperServer zk) throws IOException {
        this.self = self;
        try {
            if (self.getQuorumListenOnAllIPs()) {
                ss = new ServerSocket(self.getQuorumAddress().getPort());
            } else {
                ss = new ServerSocket();
            }
            ss.setReuseAddress(true);
            if (!self.getQuorumListenOnAllIPs()) {
                ss.bind(self.getQuorumAddress());
            }
        } catch (BindException e) {
            if (self.getQuorumListenOnAllIPs()) {
                LOG.error("Couldn't bind to port " + self.getQuorumAddress().getPort(), e);
            } else {
                LOG.error("Couldn't bind to " + self.getQuorumAddress(), e);
            }
            throw e;
        }
        this.zk = zk;
    }

    /**
     * Get string representation of a given packet type
     * @param packetType
     * @return string representing the packet type
     */
    public static String getPacketType(int packetType) {
        switch (packetType) {
            case DIFF:
                return "DIFF";
            case TRUNC:
                return "TRUNC";
            case SNAP:
                return "SNAP";
            case OBSERVERINFO:
                return "OBSERVERINFO";
            case NEWLEADER:
                return "NEWLEADER";
            case FOLLOWERINFO:
                return "FOLLOWERINFO";
            case UPTODATE:
                return "UPTODATE";
            case LEADERINFO:
                return "LEADERINFO";
            case ACKEPOCH:
                return "ACKEPOCH";
            case REQUEST:
                return "REQUEST";
            case PROPOSAL:
                return "PROPOSAL";
            case ACK:
                return "ACK";
            case COMMIT:
                return "COMMIT";
            case PING:
                return "PING";
            case REVALIDATE:
                return "REVALIDATE";
            case SYNC:
                return "SYNC";
            case INFORM:
                return "INFORM";
            default:
                return "UNKNOWN";
        }
    }

    /**
     * Returns a copy of the current learner snapshot
     */
    public List<LearnerHandler> getLearners() {
        synchronized (learners) {
            return new ArrayList<LearnerHandler>(learners);
        }
    }

    /**
     * Returns a copy of the current forwarding follower snapshot
     */
    public List<LearnerHandler> getForwardingFollowers() {
        synchronized (forwardingFollowers) {
            return new ArrayList<LearnerHandler>(forwardingFollowers);
        }
    }

    private void addForwardingFollower(LearnerHandler lh) {
        synchronized (forwardingFollowers) {
            forwardingFollowers.add(lh);
        }
    }

    /**
     * Returns a copy of the current observer snapshot
     */
    public List<LearnerHandler> getObservingLearners() {
        synchronized (observingLearners) {
            return new ArrayList<LearnerHandler>(observingLearners);
        }
    }

    private void addObserverLearnerHandler(LearnerHandler lh) {
        synchronized (observingLearners) {
            observingLearners.add(lh);
        }
    }

    synchronized public int getNumPendingSyncs() {
        return pendingSyncs.size();
    }

    /**
     * Adds peer to the leader.
     *
     * @param learner
     *                instance of learner handle
     */
    void addLearnerHandler(LearnerHandler learner) {
        synchronized (learners) {
            learners.add(learner);
        }
    }

    /**
     * Remove the learner from the learner list
     *
     * @param peer
     */
    void removeLearnerHandler(LearnerHandler peer) {
        synchronized (forwardingFollowers) {
            forwardingFollowers.remove(peer);
        }
        synchronized (learners) {
            learners.remove(peer);
        }
        synchronized (observingLearners) {
            observingLearners.remove(peer);
        }
    }

    boolean isLearnerSynced(LearnerHandler peer) {
        synchronized (forwardingFollowers) {
            return forwardingFollowers.contains(peer);
        }
    }

    /**
     * This method is main function that is called to lead
     *
     * @throws IOException
     * @throws InterruptedException
     */
    void lead() throws IOException, InterruptedException {
        // 设置Leader选举结束时间
        self.end_fle = Time.currentElapsedTime();
        long electionTimeTaken = self.end_fle - self.start_fle;
        self.setElectionTimeTaken(electionTimeTaken);
        LOG.info("LEADING - LEADER ELECTION TOOK - {}", electionTimeTaken);
        self.start_fle = 0;
        self.end_fle = 0;

        zk.registerJMX(new LeaderBean(this, zk), self.jmxLocalPeerBean);

        try {
            self.tick.set(0);
            zk.loadData();

            leaderStateSummary = new StateSummary(self.getCurrentEpoch(), zk.getLastProcessedZxid());

            // 开始线程等待Follower连接请求。Start thread that waits for connection requests from new followers.
            cnxAcceptor = new LearnerCnxAcceptor();
            cnxAcceptor.start();

            readyToStart = true;


            // 启动的Leader、Follower服务器之间协商epoch
            long epoch = getEpochToPropose(self.getId(), self.getAcceptedEpoch());

            // 此Leader任期内，起始zxid
            zk.setZxid(ZxidUtils.makeZxid(epoch, 0));

            synchronized (this) {
                lastProposed = zk.getZxid(); // 此Leader任期内，起始zxid
            }

            // Leader开始管理集群： Leader提议出一个Epoch和zxid之后，发送 NEWLEADER 消息，询问所有的Follower是否同意此Epoch和zxid
            // todo:这个 NEWLEADER 数据包，在什么时候发送的？
            newLeaderProposal.packet = new QuorumPacket(NEWLEADER, zk.getZxid(), // zk.getZxid()是此Leader任期内，起始zxid
                                                        null, null);


            if ((newLeaderProposal.packet.getZxid() & 0xffffffffL) != 0) {
                LOG.info("NEWLEADER proposal has Zxid of " + Long.toHexString(newLeaderProposal.packet.getZxid()));
            }

            // Leader节点等待learner节点ack
            // 【注意：这里如果集群中server不达到法定人数，就会wait；如果超时就会抛出异常!!!】
            waitForEpochAck(self.getId(), leaderStateSummary);

            self.setCurrentEpoch(epoch);

            try {
                // 处理 NEWLEADER ACK ，wait直到Leader服务器收到足够的服务器 ACK。
                // 也就是等待过半的server完成数据同步。
                waitForNewLeaderAck(self.getId(), zk.getZxid(), LearnerType.PARTICIPANT);
            } catch (InterruptedException e) {
                shutdown("Waiting for a quorum of followers, only synced with sids: [ " + getSidSetString(newLeaderProposal.ackSet) + " ]");
                HashSet<Long> followerSet = new HashSet<Long>();
                for (LearnerHandler f : learners) {
                    followerSet.add(f.getSid());
                }

                if (self.getQuorumVerifier().containsQuorum(followerSet)) {
                    LOG.warn("Enough followers present. " + "Perhaps the initTicks need to be increased.");
                }
                Thread.sleep(self.tickTime);
                self.tick.incrementAndGet();
                return;
            }

            // 数据同步完毕，Leader服务器启动。
            startZkServer();

            /**
             * WARNING: do not use this for anything other than QA testing
             * on a real cluster. Specifically to enable verification that quorum
             * can handle the lower 32bit roll-over issue identified in
             * ZOOKEEPER-1277. Without this option it would take a very long
             * time (on order of a month say) to see the 4 billion writes
             * necessary to cause the roll-over to occur.
             *
             * This field allows you to override the zxid of the server. Typically
             * you'll want to set it to something like 0xfffffff0 and then
             * start the quorum, run some operations and see the re-election.
             */
            String initialZxid = System.getProperty("zookeeper.testingonly.initialZxid");
            if (initialZxid != null) {
                long zxid = Long.parseLong(initialZxid);
                zk.setZxid((zk.getZxid() & 0xffffffff00000000L) | zxid);
            }

            if (!System.getProperty("zookeeper.leaderServes", "yes").equals("no")) {
                self.cnxnFactory.setZooKeeperServer(zk);
            }
            // Everything is a go, simply start counting the ticks
            // WARNING: I couldn't find any wait statement on a synchronized
            // block that would be notified by this notifyAll() call, so
            // I commented it out
            //synchronized (this) {
            //    notifyAll();
            //}
            // We ping twice a tick, so we only update the tick every other iteration
            boolean tickSkip = true;

            while (true) {
                // 每次循环，间隔1000毫秒。
                Thread.sleep(self.tickTime / 2);

                if (!tickSkip) {
                    self.tick.incrementAndGet();
                }
                HashSet<Long> syncedSet = new HashSet<Long>();

                // lock on the followers when we use it.
                syncedSet.add(self.getId());

                for (LearnerHandler f : getLearners()) {
                    // syncedSet 为了检查Follower服务器是否达到法定人数。
                    if (f.synced() && f.getLearnerType() == LearnerType.PARTICIPANT) {
                        syncedSet.add(f.getSid());
                    }
                    f.ping();
                }

                // 检查Leader服务器运行状态
                if (!this.isRunning()) {
                    shutdown("Unexpected internal error");
                    return;
                }

                // 检查当前Follower服务器是否达到法定人数
                if (!tickSkip && !self.getQuorumVerifier().containsQuorum(syncedSet)) {
                    // 没有达到法定人数，则关闭Leader服务器。
                    shutdown("Not sufficient followers synced, only synced with sids: [ " + getSidSetString(syncedSet) + " ]");
                    return;
                }
                // 每两次tick，给Learner服务器发送一次ping请求。
                tickSkip = !tickSkip;
            }
        } finally {
            zk.unregisterJMX(this);
        }
    }

    /**
     * Close down all the LearnerHandlers
     */
    void shutdown(String reason) {
        LOG.info("Shutting down");

        if (isShutdown) {
            return;
        }

        LOG.info("Shutdown called", new Exception("shutdown Leader! reason: " + reason));

        if (cnxAcceptor != null) {
            cnxAcceptor.halt();
        }

        // NIO should not accept conenctions
        self.cnxnFactory.setZooKeeperServer(null);
        try {
            ss.close();
        } catch (IOException e) {
            LOG.warn("Ignoring unexpected exception during close", e);
        }
        // clear all the connections
        self.cnxnFactory.closeAll();
        // shutdown the previous zk
        if (zk != null) {
            zk.shutdown();
        }
        synchronized (learners) {
            for (Iterator<LearnerHandler> it = learners.iterator(); it.hasNext(); ) {
                LearnerHandler f = it.next();
                it.remove();
                f.shutdown();
            }
        }
        isShutdown = true;
    }

    /**
     * Keep a count of acks that are received by the leader for a particular
     * proposal
     *
     * @param zxid
     *                the zxid of the proposal sent out
     * @param followerAddr
     */
    synchronized public void processAck(long sid, long zxid, SocketAddress followerAddr) {
        if (LOG.isTraceEnabled()) {
            LOG.trace("Ack zxid: 0x{}", Long.toHexString(zxid));
            for (Proposal p : outstandingProposals.values()) {
                long packetZxid = p.packet.getZxid();
                LOG.trace("outstanding proposal: 0x{}", Long.toHexString(packetZxid));
            }
            LOG.trace("outstanding proposals all");
        }

        if ((zxid & 0xffffffffL) == 0) {
            /*
             * We no longer process NEWLEADER ack by this method. However,
             * the learner sends ack back to the leader after it gets UPTODATE
             * so we just ignore the message.
             */
            return;
        }

        if (outstandingProposals.size() == 0) {
            if (LOG.isDebugEnabled()) {
                LOG.debug("outstanding is 0");
            }
            return;
        }
        if (lastCommitted >= zxid) {
            if (LOG.isDebugEnabled()) {
                LOG.debug("proposal has already been committed, pzxid: 0x{} zxid: 0x{}", Long.toHexString(lastCommitted), Long.toHexString(zxid));
            }
            // The proposal has already been committed
            return;
        }
        Proposal p = outstandingProposals.get(zxid);
        if (p == null) {
            LOG.warn("Trying to commit future proposal: zxid 0x{} from {}", Long.toHexString(zxid), followerAddr);
            return;
        }

        p.ackSet.add(sid);
        if (LOG.isDebugEnabled()) {
            LOG.debug("Count for zxid: 0x{} is {}", Long.toHexString(zxid), p.ackSet.size());
        }
        if (self.getQuorumVerifier().containsQuorum(p.ackSet)) {
            if (zxid != lastCommitted + 1) {
                LOG.warn("Commiting zxid 0x{} from {} not first!", Long.toHexString(zxid), followerAddr);
                LOG.warn("First is 0x{}", Long.toHexString(lastCommitted + 1));
            }
            outstandingProposals.remove(zxid);
            if (p.request != null) {
                toBeApplied.add(p);
            }

            if (p.request == null) {
                LOG.warn("Going to commmit null request for proposal: {}", p);
            }
            commit(zxid);
            inform(p);
            zk.commitProcessor.commit(p.request);
            if (pendingSyncs.containsKey(zxid)) {
                for (LearnerSyncRequest r : pendingSyncs.remove(zxid)) {
                    sendSync(r);
                }
            }
        }
    }

    /**
     * send a packet to all the followers ready to follow
     *
     * @param qp
     *                the packet to be sent
     */
    void sendPacket(QuorumPacket qp) {
        synchronized (forwardingFollowers) {
            for (LearnerHandler f : forwardingFollowers) {
                f.queuePacket(qp);
            }
        }
    }

    /**
     * send a packet to all observers
     */
    void sendObserverPacket(QuorumPacket qp) {
        for (LearnerHandler f : getObservingLearners()) {
            f.queuePacket(qp);
        }
    }

    /**
     * Create a commit packet and send it to all the members of the quorum
     *
     * @param zxid
     */
    public void commit(long zxid) {
        synchronized (this) {
            lastCommitted = zxid;
        }
        QuorumPacket qp = new QuorumPacket(Leader.COMMIT, zxid, null, null);
        sendPacket(qp);
    }

    /**
     * Create an inform packet and send it to all observers.
     * @param zxid
     * @param proposal
     */
    public void inform(Proposal proposal) {
        QuorumPacket qp = new QuorumPacket(Leader.INFORM, proposal.request.zxid, proposal.packet.getData(), null);
        sendObserverPacket(qp);
    }

    /**
     * Returns the current epoch of the leader.
     *
     * @return
     */
    public long getEpoch() {
        return ZxidUtils.getEpochFromZxid(lastProposed);
    }

    /**
     * create a proposal and send it out to all the members
     *
     * @param request
     * @return the proposal that is queued to send to all the members
     */
    public Proposal propose(Request request) throws XidRolloverException {
        /**
         * Address the rollover issue. All lower 32bits set indicate a new leader
         * election. Force a re-election instead. See ZOOKEEPER-1277
         */
        if ((request.zxid & 0xffffffffL) == 0xffffffffL) {
            String msg = "zxid lower 32 bits have rolled over, forcing re-election, and therefore new epoch start";
            shutdown(msg);
            throw new XidRolloverException(msg);
        }

        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        BinaryOutputArchive boa = BinaryOutputArchive.getArchive(baos);
        try {
            request.hdr.serialize(boa, "hdr");
            if (request.txn != null) {
                request.txn.serialize(boa, "txn");
            }
            baos.close();
        } catch (IOException e) {
            LOG.warn("This really should be impossible", e);
        }
        QuorumPacket pp = new QuorumPacket(Leader.PROPOSAL, request.zxid, baos.toByteArray(), null);

        Proposal p = new Proposal();
        p.packet = pp;
        p.request = request;
        synchronized (this) {
            if (LOG.isDebugEnabled()) {
                LOG.debug("Proposing:: " + request);
            }

            lastProposed = p.packet.getZxid();
            outstandingProposals.put(lastProposed, p);
            sendPacket(pp);
        }
        return p;
    }

    /**
     * Process sync requests
     *
     * @param r the request
     */

    synchronized public void processSync(LearnerSyncRequest r) {
        if (outstandingProposals.isEmpty()) {
            sendSync(r);
        } else {
            List<LearnerSyncRequest> l = pendingSyncs.get(lastProposed);
            if (l == null) {
                l = new ArrayList<LearnerSyncRequest>();
            }
            l.add(r);
            pendingSyncs.put(lastProposed, l);
        }
    }

    /**
     * Sends a sync message to the appropriate server
     *
     * @param f
     * @param r
     */

    public void sendSync(LearnerSyncRequest r) {
        QuorumPacket qp = new QuorumPacket(Leader.SYNC, 0, null, null);
        r.fh.queuePacket(qp);
    }

    /**
     * lets the leader know that a follower is capable of following and is done
     * syncing
     *
     * @param handler handler of the follower
     * @return last proposed zxid
     */
    synchronized public long startForwarding(LearnerHandler handler, long lastSeenZxid) {
        // Queue up any outstanding requests enabling the receipt of
        // new requests
        if (lastProposed > lastSeenZxid) {
            for (Proposal p : toBeApplied) {
                if (p.packet.getZxid() <= lastSeenZxid) {
                    continue;
                }
                handler.queuePacket(p.packet);
                // Since the proposal has been committed we need to send the
                // commit message also
                QuorumPacket qp = new QuorumPacket(Leader.COMMIT, p.packet.getZxid(), null, null);
                handler.queuePacket(qp);
            }
            // Only participant need to get outstanding proposals
            if (handler.getLearnerType() == LearnerType.PARTICIPANT) {
                List<Long> zxids = new ArrayList<Long>(outstandingProposals.keySet());
                Collections.sort(zxids);
                for (Long zxid : zxids) {
                    if (zxid <= lastSeenZxid) {
                        continue;
                    }
                    handler.queuePacket(outstandingProposals.get(zxid).packet);
                }
            }
        }
        if (handler.getLearnerType() == LearnerType.PARTICIPANT) {
            addForwardingFollower(handler);
        } else {
            addObserverLearnerHandler(handler);
        }

        return lastProposed;
    }

    /**
     * 启动的Leader、Follower服务器之间协商epoch；如果启动的server无法组成法定人数，则wait
     *
     * 注意：这里的wait逻辑：
     * 1、Leader在启动之后，会根据上一个Leader周期公认的Epoch，来调用此方法。
     * 2、Leader服务器在接收到Learner服务器的连接请求后，(LearnerHandler#run方法中)接收FOLLOWERINFO或者OBSERVERINFO请求，并根据其sid和最后epoch，调用此方法。
     * 3、如果连接的Follower服务器数量未达到法定人数，则wait。如果超时仍未完成waitingForNewEpoch，则抛出异常。
     * 4、如果达到法定人数，则结束此方法，返回epoch：
     *    这个epoch表示：当前Leader和Learner之间共同商议出的一个当前Leader周期。
     *
     * @param sid 添加到 connectingFollowers 中。
     * @param lastAcceptedEpoch
     * @return 当前启动的server写上出的新Epoch
     * @throws InterruptedException
     * @throws IOException
     */
    public long getEpochToPropose(long sid, long lastAcceptedEpoch) throws InterruptedException, IOException {
        synchronized (connectingFollowers) {
            if (!waitingForNewEpoch) {
                return epoch;
            }
            // 在集群中，最后公认的Epoch自增1为当前Leader的Epoch
            if (lastAcceptedEpoch >= epoch) {
                epoch = lastAcceptedEpoch + 1;
            }
            connectingFollowers.add(sid);

            // 判断是否达到法定人数
            if (connectingFollowers.contains(self.getId()) && self.getQuorumVerifier().containsQuorum(connectingFollowers)) {
                waitingForNewEpoch = false;
                self.setAcceptedEpoch(epoch);
                connectingFollowers.notifyAll();
            } else {
                long start = Time.currentElapsedTime();
                long cur = start;
                long end = start + self.getInitLimit() * self.getTickTime();
                while (waitingForNewEpoch && cur < end) {
                    connectingFollowers.wait(end - cur);
                    cur = Time.currentElapsedTime();
                }
                if (waitingForNewEpoch) {
                    throw new InterruptedException("Timeout while waiting for epoch from quorum");
                }
            }
            return epoch;
        }
    }

    /**
     * Leader节点在这里调用此方法的目的：如果只有leader调用此方法，则containsQuorum()方法判断条件一定无法成立，则一定会执行wait等待。
     *                                  所以主要目的就是为了把 self.sid 添加 到 electingFollowers 中，并等待
     * @param id
     * @param ss
     * @throws IOException
     * @throws InterruptedException
     */
    public void waitForEpochAck(long id, StateSummary ss) throws IOException, InterruptedException {
        synchronized (electingFollowers) {
            if (electionFinished) {
                return;
            }

            // 如果等于-1，就不会把当前Learner sid添加到electingFollowers中了。
            if (ss.getCurrentEpoch() != -1) {
                if (ss.isMoreRecentThan(leaderStateSummary)) {
                    throw new IOException("Follower is ahead of the leader, leader summary: " + leaderStateSummary.getCurrentEpoch() + " (current epoch), " + leaderStateSummary.getLastZxid() + " (last zxid)");
                }
                electingFollowers.add(id);
            }

            // 判断electingFollowers中的sid能否构成法定人数
            if (electingFollowers.contains(self.getId()) && self.getQuorumVerifier().containsQuorum(electingFollowers)) {
                // 如果可以，选举结束
                electionFinished = true;
                electingFollowers.notifyAll();
            } else {
                // 如果sid数量无法构成法定人数，则等待一定时间。等到事件超时后，会报错。
                long start = Time.currentElapsedTime();
                long cur = start;
                // 在Learner与Leader进行数据同步阶段，wait超时时间。
                long end = start + self.getInitLimit() * self.getTickTime();
                while (!electionFinished && cur < end) {
                    electingFollowers.wait(end - cur);
                    cur = Time.currentElapsedTime();
                }

                if (!electionFinished) {
                    throw new InterruptedException("Timeout while waiting for epoch to be acked by quorum");
                }
            }
        }
    }

    /**
     * Return a list of sid in set as string
     */
    private String getSidSetString(Set<Long> sidSet) {
        StringBuilder sids = new StringBuilder();
        Iterator<Long> iter = sidSet.iterator();
        while (iter.hasNext()) {
            sids.append(iter.next());
            if (!iter.hasNext()) {
                break;
            }
            sids.append(",");
        }
        return sids.toString();
    }

    /**
     * Start up Leader ZooKeeper server and initialize zxid to the new epoch
     */
    private synchronized void startZkServer() {
        // Update lastCommitted and Db's zxid to a value representing the new epoch
        lastCommitted = zk.getZxid();
        LOG.info("Have quorum of supporters, sids: [ " + getSidSetString(newLeaderProposal.ackSet) + " ]; starting up and setting last processed zxid: 0x{}", Long.toHexString(zk.getZxid()));
        zk.startup();
        /*
         * Update the election vote here to ensure that all members of the
         * ensemble report the same vote to new servers that start up and
         * send leader election notifications to the ensemble.
         *
         * @see https://issues.apache.org/jira/browse/ZOOKEEPER-1732
         */
        self.updateElectionVote(getEpoch());

        zk.getZKDatabase().setlastProcessedZxid(zk.getZxid());
    }

    /**
     * 处理 {@link Leader#NEWLEADER} ACK ，wait直到Leader服务器收到足够的服务器 ACK。
     *
     * @param sid
     * @param zxid
     * @param learnerType
     * @throws InterruptedException
     */
    public void waitForNewLeaderAck(long sid, long zxid, LearnerType learnerType) throws InterruptedException {

        synchronized (newLeaderProposal.ackSet) {
            if (quorumFormed) {
                return;
            }

            long currentZxid = newLeaderProposal.packet.getZxid();
            if (zxid != currentZxid) {
                LOG.error("NEWLEADER ACK from sid: " + sid + " is from a different epoch - current 0x" + Long.toHexString(currentZxid) + " receieved 0x" + Long.toHexString(zxid));
                return;
            }

            if (learnerType == LearnerType.PARTICIPANT) {
                newLeaderProposal.ackSet.add(sid);
            }

            if (self.getQuorumVerifier().containsQuorum(newLeaderProposal.ackSet)) {
                quorumFormed = true;
                newLeaderProposal.ackSet.notifyAll();
            } else {
                long start = Time.currentElapsedTime();
                long cur = start;
                long end = start + self.getInitLimit() * self.getTickTime();
                while (!quorumFormed && cur < end) {
                    newLeaderProposal.ackSet.wait(end - cur);
                    cur = Time.currentElapsedTime();
                }
                if (!quorumFormed) {
                    throw new InterruptedException("Timeout while waiting for NEWLEADER to be acked by quorum");
                }
            }
        }
    }

    private boolean isRunning() {
        return self.isRunning() && zk.isRunning();
    }

    static public class Proposal {
        public QuorumPacket packet;

        public HashSet<Long> ackSet = new HashSet<Long>();

        public Request request;

        @Override
        public String toString() {
            return packet.getType() + ", " + packet.getZxid() + ", " + request;
        }
    }

    static class ToBeAppliedRequestProcessor implements RequestProcessor {
        private RequestProcessor next;

        private ConcurrentLinkedQueue<Proposal> toBeApplied;

        /**
         * This request processor simply maintains the toBeApplied list. For
         * this to work next must be a FinalRequestProcessor and
         * FinalRequestProcessor.processRequest MUST process the request
         * synchronously!
         *
         * @param next
         *                a reference to the FinalRequestProcessor
         */
        ToBeAppliedRequestProcessor(RequestProcessor next, ConcurrentLinkedQueue<Proposal> toBeApplied) {
            if (!(next instanceof FinalRequestProcessor)) {
                throw new RuntimeException(ToBeAppliedRequestProcessor.class.getName() + " must be connected to " + FinalRequestProcessor.class.getName() + " not " + next.getClass().getName());
            }
            this.toBeApplied = toBeApplied;
            this.next = next;
        }

        /*
         * (non-Javadoc)
         *
         * @see org.apache.zookeeper.server.RequestProcessor#processRequest(org.apache.zookeeper.server.Request)
         */
        @Override
        public void processRequest(Request request) throws RequestProcessorException {
            // request.addRQRec(">tobe");
            next.processRequest(request);
            Proposal p = toBeApplied.peek();
            if (p != null && p.request != null
                    && p.request.zxid == request.zxid) {
                toBeApplied.remove();
            }
        }

        /*
         * (non-Javadoc)
         *
         * @see org.apache.zookeeper.server.RequestProcessor#shutdown()
         */
        @Override
        public void shutdown() {
            LOG.info("Shutting down");
            next.shutdown();
        }
    }

    @SuppressWarnings("serial")
    public static class XidRolloverException extends Exception {
        public XidRolloverException(String message) {
            super(message);
        }
    }

    class LearnerCnxAcceptor extends ZooKeeperThread {
        private volatile boolean stop = false;

        public LearnerCnxAcceptor() {
            super("LearnerCnxAcceptor-" + ss.getLocalSocketAddress());
        }

        @Override
        public void run() {
            try {
                while (!stop) {
                    try {
                        Socket s = ss.accept();
                        // start with the initLimit, once the ack is processed
                        // in LearnerHandler switch to the syncLimit
                        s.setSoTimeout(self.tickTime * self.initLimit);
                        s.setTcpNoDelay(nodelay);

                        BufferedInputStream is = new BufferedInputStream(s.getInputStream());
                        LearnerHandler fh = new LearnerHandler(s, is, Leader.this);
                        fh.start();
                    } catch (SocketException e) {
                        if (stop) {
                            LOG.info("exception while shutting down acceptor: " + e);

                            // When Leader.shutdown() calls ss.close(), // the call to accept throws an exception.
                            // We catch and set stop to true.
                            stop = true;
                        } else {
                            throw e;
                        }
                    } catch (SaslException e) {
                        LOG.error("Exception while connecting to quorum learner", e);
                    }
                }
            } catch (Exception e) {
                LOG.warn("Exception while accepting follower", e);
            }
        }

        public void halt() {
            stop = true;
        }
    }
}
